package net.xdclass.xdclassmq.controller;

import net.xdclass.xdclassmq.config.JmsConfig;
import net.xdclass.xdclassmq.domain.ProductOrder;
import net.xdclass.xdclassmq.jms.producer.PayProducer;
import net.xdclass.xdclassmq.jms.producer.TransactionProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.HashMap;
import java.util.List;

@RestController
public class PayController {

    @Autowired
    private PayProducer payProducer;

    @Autowired
    private JmsConfig jmsConfig;

    @Autowired
    private TransactionProducer transactionProducer;

    /**
     * 同步发送
     * @param text
     * @return
     * @throws Exception
     */
    @RequestMapping("/api/v1/sync")
    public Object sync(String text) throws Exception {
        Message message = new Message(jmsConfig.getPayTopic(),"taga","6688",("hello xdclass rocketmq = " + text).getBytes());
        SendResult sendResult = payProducer.getProducer().send(message);
        /*if(SendStatus.SLAVE_NOT_AVAILABLE.equals(sendResult.getSendStatus())){
            System.out.println("子节点不存在，同步子节点消息出错");
        }*/
        System.out.println(sendResult);

        return new HashMap<>();
    }

    /**
     * 异步发送
     * 使用在一些对响应时间敏感的业务场景，即发送端不能容忍长时间地等待Broker的响应
     * @param text
     * @return
     * @throws Exception
     */
    @RequestMapping("/api/v1/async")
    public Object async(String text) throws Exception {
        Message message = new Message(jmsConfig.getPayTopic(),"taga","6688",("hello xdclass rocketmq = " + text).getBytes());

        payProducer.getProducer().send(message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.printf("发送结果=%s,msg=%s",sendResult.getSendStatus(),sendResult.toString());
            }

            @Override
            public void onException(Throwable throwable) {
                //补偿机制，根据业务情况进行使用，看是否进行重试
                throwable.printStackTrace();
            }
        });

        return new HashMap<>();
    }

    /**
     * 单向发送
     * 无需等待响应，主要是日志收集，适用于某些耗时非常短，但对可靠性并不高的场景，也就是LogServer，
     * 只负责发送消息，不等待服务器回应且没有回调函数触发，即只发送请求不等待应答
     * @param text
     * @return
     * @throws Exception
     */
    @RequestMapping("/api/v1/oneWay")
    public Object oneWay(String text) throws Exception {
        Message message = new Message(jmsConfig.getPayTopic(),"taga","6688",("hello xdclass rocketmq = " + text).getBytes());

        payProducer.getProducer().sendOneway(message);

        return new HashMap<>();
    }

    /**
     * 延迟消费
     * @param text
     * @return
     * @throws Exception
     */
    @RequestMapping("/api/v1/delay")
    public Object delay(String text) throws Exception {
        Message message = new Message(jmsConfig.getPayTopic(),"taga","6688",("hello xdclass rocketmq = " + text).getBytes());

        //消息被消费的时间延迟级别：1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
        //级别2代表5s以后被消费
        message.setDelayTimeLevel(2);
        SendResult sendResult = payProducer.getProducer().send(message);
        System.out.println(sendResult);

        return new HashMap<>();
    }


    /**
     * 同步发送
     * 顺序消费，分摊负载
     * 生产消息使用MessageQueueSelector投递到Topic下指定的queue
     * @param text
     * @return
     * @throws Exception
     */
    @RequestMapping("/api/v1/messageQueue")
    public Object MessageQueue(String text) throws Exception {
        Message message = new Message(jmsConfig.getPayTopic(),"taga","6688",("hello xdclass rocketmq = " + text).getBytes());

        //同步发送
        //发送消息到队列0中
        SendResult sendResult = payProducer.getProducer().send(message,new MessageQueueSelector(){
            @Override
            public MessageQueue select(List<MessageQueue> list, Message message, Object o) {
                int queueNum = Integer.valueOf(o.toString());
                return list.get(queueNum);
            }
        },5);
        System.out.println(sendResult);

        return new HashMap<>();
    }


    /**
     * 异步发送
     * 顺序消费，分摊负载
     * 生产消息使用MessageQueueSelector投递到Topic下指定的queue
     * @param text
     * @return
     * @throws Exception
     */
    @RequestMapping("/api/v1/async/messageQueue")
    public Object asyncMessageQueue(String text) throws Exception {
        Message message = new Message(jmsConfig.getPayTopic(),"taga","6688",("hello xdclass rocketmq = " + text).getBytes());

        //异步发送
        //发送消息到队列0中
        payProducer.getProducer().send(message, (list, message1, o) -> {
            int queueNum = Integer.valueOf(o.toString());
            return list.get(queueNum);
        }, 6, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.printf("发送结果=%s,msg=%s",sendResult.getSendStatus(),sendResult.toString());
                System.out.println();
            }

            @Override
            public void onException(Throwable throwable) {
                throwable.printStackTrace();
            }
        });

        return new HashMap<>();
    }


    /**
     * 根据orderId发送到指定queue
     * @param text
     * @return
     * @throws Exception
     */
    @RequestMapping("/api/v1/orderMessage")
    public Object orderMessage(String text) throws Exception {
        List<ProductOrder> list = ProductOrder.getOrderList();

        /*list.forEach(order -> {
            Message message = new Message(jmsConfig.getOrderlyTopic(),"",
                    order.getOrderId()+"",order.toString().getBytes());
            SendResult sendResult = payProducer.getProducer()
                    .send(message, (mqs, msg, o) -> {
                        Long id = (Long) o;
                        long index = id % mqs.size();
                        return mqs.get((int)index);
                    },order.getOrderId());
        });*/
        for(ProductOrder order : list){
            Message message = new Message(jmsConfig.getOrderlyTopic(),"",
                    order.getOrderId()+"",order.toString().getBytes());
            SendResult sendResult = payProducer.getProducer()
                    .send(message, (mqs, msg, o) -> {
                        Long id = (Long) o;
                        long index = id % mqs.size();
                        return mqs.get((int)index);
                    },order.getOrderId());
            System.out.printf("发送结果=%s，sendResult=%s，orderId=%s，type=%s\n",
                    sendResult.getSendStatus(),sendResult.toString(),
                    order.getOrderId(),order.getType());
        }

        return new HashMap<>();
    }


    /**
     * 发送指定tag的消息
     * @param tag
     * @return
     * @throws Exception
     */
    @RequestMapping("/api/v1/tag")
    public Object tag(String tag) throws Exception {
        Message message = new Message(jmsConfig.getPayTopic(),tag,"",tag.getBytes());
        SendResult sendResult = payProducer.getProducer().send(message);
        System.out.println(sendResult);

        return new HashMap<>();
    }

    /**
     * 测试 根据筛选条件订阅消息
     * @param amount
     * @return
     * @throws Exception
     */
    @RequestMapping("/api/v1/amount")
    public Object tag(int amount) throws Exception {
        Message message = new Message(jmsConfig.getPayTopic(),"","","123".getBytes());
        message.putUserProperty("amount",amount+"");
        SendResult sendResult = payProducer.getProducer().send(message);
        System.out.println(sendResult);

        return new HashMap<>();
    }

    /**
     * 测试分布式事务消息
     * @param tag
     * @param otherParam
     * @return
     * @throws Exception
     */
    @RequestMapping("/api/v1/trans")
    public Object trans(String tag,String otherParam) throws Exception {
        Message message = new Message(jmsConfig.getPayTopic(),tag,tag+"_key",tag.getBytes());
        TransactionSendResult sendResult = transactionProducer.getProducer().sendMessageInTransaction(message,otherParam);
        System.out.println(sendResult);

        return new HashMap<>();
    }

    @GetMapping("hello")
    public Object hello(){
        System.out.println("hello world");
        return new HashMap<>();
    }
}
